package com.yvon.maple.service.msg.core.support.action;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.google.common.base.Throwables;
import com.yvon.boot.common.result.ResponseData;
import com.yvon.boot.common.result.ResultCode;
import com.yvon.maple.service.msg.core.common.model.send.SendTaskModel;
import com.yvon.maple.service.msg.core.support.pipeline.BusinessProcess;
import com.yvon.maple.service.msg.core.support.pipeline.ProcessContext;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

/**
 * 将消息发送到MQ
 *
 * @author : Yvon
 * @since : 2021-12-23
 */
@Slf4j
@Service
public class SendMqAction implements BusinessProcess {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    @Value("${maple.topic.name}")
    private String topicName;

    @Override
    public void process(ProcessContext context) {
        SendTaskModel sendTaskModel = (SendTaskModel) context.getProcessModel();

        try {
            kafkaTemplate.send(topicName, JSON.toJSONString(sendTaskModel.getTaskInfo(),
                    SerializerFeature.WriteClassName));
        } catch (Exception e) {
            context.setNeedBreak(true).setResponse(ResponseData.failed(ResultCode.SERVICE_ERROR));
            log.error("send kafka fail! e:{},params:{}", Throwables.getStackTraceAsString(e)
                    , JSON.toJSONString(sendTaskModel.getTaskInfo().get(0)));
        }
    }
}
